package com.gzx.plugin.tjzy.modular.call.receipt.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fhs.trans.service.impl.TransService;
import com.google.common.collect.Lists;
import com.gzx.plugin.tjzy.modular.caffeine.UserCaffeine;
import com.gzx.plugin.tjzy.modular.call.ht.properties.CallInProperties;
import com.gzx.plugin.tjzy.modular.call.predict.entity.GzxPredictCallDetail;
import com.gzx.plugin.tjzy.modular.call.predict.service.GzxPredictCallDbService;
import com.gzx.plugin.tjzy.modular.call.receipt.bo.GzxHtCallReceiptQueryBo;
import com.gzx.plugin.tjzy.modular.call.receipt.consts.GzxHtCallTypeConstant;
import com.gzx.plugin.tjzy.modular.call.receipt.dto.GzxCallCheckDto;
import com.gzx.plugin.tjzy.modular.call.receipt.entity.GzxHtCallReceiptBusiness;
import com.gzx.plugin.tjzy.modular.call.receipt.mapper.GzxHtCallReceiptBusinessMapper;
import com.gzx.plugin.tjzy.modular.call.receipt.service.GzxHtCallReceiptBusinessService;
import com.gzx.plugin.tjzy.modular.call.receipt.vo.GzxHtCallReceiptBusinessVo;
import com.gzx.plugin.tjzy.modular.call.receipt.vo.GzxHtCallReportVo;
import com.gzx.plugin.tjzy.modular.gzxcase.entity.GzxCase;
import com.gzx.plugin.tjzy.modular.gzxcase.service.GzxCaseDbService;
import com.gzx.plugin.tjzy.modular.liaison.entity.GzxDebtorLiaison;
import com.gzx.plugin.tjzy.modular.liaison.service.GzxDebtorLiaisonService;
import com.gzx.plugin.tjzy.modular.work.entity.GzxMediatorWork;
import com.gzx.plugin.tjzy.modular.work.service.GzxMediatorWorkService;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Service;
import vip.xiaonuo.common.consts.GzxTenantConstant;
import vip.xiaonuo.common.consts.IntConstant;
import vip.xiaonuo.common.exception.CommonException;
import vip.xiaonuo.common.kafka.KafkaMessageProducer;
import vip.xiaonuo.common.kafka.KafkaTopicConstant;
import vip.xiaonuo.common.mail.WechatMailUtil;
import vip.xiaonuo.common.mail.dto.MailRequestDto;
import vip.xiaonuo.common.pojo.ExcelCalleeVo;
import vip.xiaonuo.common.tenant.TenantContext;
import vip.xiaonuo.common.util.*;
import vip.xiaonuo.common.wechat.EmailExcelUtils;

import javax.mail.MessagingException;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import static org.apache.commons.lang.StringEscapeUtils.escapeSql;

/**
 * @author zjt
 * @description
 * @date 2025/3/28 18:30
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class GzxHtCallReceiptBusinessServiceImpl implements GzxHtCallReceiptBusinessService {

    private final GzxHtCallReceiptBusinessMapper htCallReceiptBusinessMapper;

    private final CallInProperties callInProperties;

    private final GzxMediatorWorkService mediatorWorkService;

    private final GzxDebtorLiaisonService debtorLiaisonService;

    private final GzxPredictCallDbService predictCallDbService;

    private final UserCaffeine userCaffeine;

    private final GzxCaseDbService caseDbService;

    private final WechatMailUtil wechatMailUtil;

    private final TransService transService;

    private final KafkaMessageProducer kafkaMessageProducer;

    /**
     * 分页查询话单记录
     *
     * @param htCallReceiptQueryBo
     * @return
     */
    @Override
    @DS("tjzy-async")
    public Page<GzxHtCallReceiptBusinessVo> pageList(GzxHtCallReceiptQueryBo htCallReceiptQueryBo) {
        Page<GzxHtCallReceiptBusinessVo> businessVoPage = htCallReceiptBusinessMapper.selectVoPage(new Page<>(htCallReceiptQueryBo.getCurrent(), htCallReceiptQueryBo.getSize()), Wrappers.<GzxHtCallReceiptBusiness>lambdaQuery()
                .eq(htCallReceiptQueryBo.getCaseId() != null, GzxHtCallReceiptBusiness::getCaseId, htCallReceiptQueryBo.getCaseId())
                .eq(StringUtils.isNotEmpty(htCallReceiptQueryBo.getCallId()), GzxHtCallReceiptBusiness::getCallId, htCallReceiptQueryBo.getCallId())
                .between(htCallReceiptQueryBo.getCaseId() == null, GzxHtCallReceiptBusiness::getCreatedTime, htCallReceiptQueryBo.getBeginTime(), htCallReceiptQueryBo.getEndTime())
                .eq(StringUtils.isNotBlank(htCallReceiptQueryBo.getCaller()), GzxHtCallReceiptBusiness::getCaller, htCallReceiptQueryBo.getCaller())
                .eq(StringUtils.isNotBlank(htCallReceiptQueryBo.getCallee()), GzxHtCallReceiptBusiness::getCallee, CommonCryptogramUtil.doSm1AesEncrypt(htCallReceiptQueryBo.getCallee()))
                .eq(StringUtils.isNotBlank(htCallReceiptQueryBo.getUserName()), GzxHtCallReceiptBusiness::getAgentNum, htCallReceiptQueryBo.getUserName())
                .gt(htCallReceiptQueryBo.getCallDuration() != null, GzxHtCallReceiptBusiness::getCallDuration, htCallReceiptQueryBo.getCallDuration())
                .eq(StringUtils.isNotBlank(htCallReceiptQueryBo.getCallType()), GzxHtCallReceiptBusiness::getCallType, htCallReceiptQueryBo.getCallType())
                .eq(htCallReceiptQueryBo.getDeptId() != null, GzxHtCallReceiptBusiness::getDeptId, htCallReceiptQueryBo.getDeptId())
                .orderByDesc(GzxHtCallReceiptBusiness::getId));
        if (CollectionUtil.isNotEmpty(businessVoPage.getRecords())) {
            Map<String, cn.hutool.json.JSONObject> userMap = userCaffeine.getUserMap();
            businessVoPage.getRecords().forEach(s -> {
                if (StringUtils.isNotBlank(s.getAgentNum()) && userMap.containsKey(s.getAgentNum())) {
                    s.setUserName(userMap.get(s.getAgentNum()).getStr("name"));
                }
            });
        }
        return businessVoPage;
    }

    /**
     * 获取坐席工作报表
     *
     * @param htCallReceiptQueryBo
     * @return
     */
    @Override
    @DS("tjzy-async")
    public Page<GzxHtCallReportVo> pageCallReport(GzxHtCallReceiptQueryBo htCallReceiptQueryBo) {
        if (htCallReceiptQueryBo.getDeptId() == null || StringUtils.isEmpty(htCallReceiptQueryBo.getBeginTime())) {
            throw new CommonException("时间和部门都不允许为空");
        }
        Page<GzxHtCallReportVo> callReportVoPage = htCallReceiptBusinessMapper.pageCallReport(new Page<>(htCallReceiptQueryBo.getCurrent(), htCallReceiptQueryBo.getSize()), htCallReceiptQueryBo.getBeginTime(), htCallReceiptQueryBo.getDeptId());
        if (CollectionUtil.isNotEmpty(callReportVoPage.getRecords())) {
            setCallReport(callReportVoPage.getRecords());
        }
        return callReportVoPage;
    }

    /**
     * 导出数据
     *
     * @param htCallReceiptQueryBo
     * @param response
     * @return
     */
    @Override
    @DS("tjzy-async")
    public void exportCallReport(GzxHtCallReceiptQueryBo htCallReceiptQueryBo, HttpServletResponse response) {
        if (htCallReceiptQueryBo.getDeptId() == null || StringUtils.isEmpty(htCallReceiptQueryBo.getBeginTime())) {
            throw new CommonException("时间和部门都不允许为空");
        }
        List<GzxHtCallReportVo> callReportVoList = htCallReceiptBusinessMapper.pageCallReport(new Page<>(1, IntConstant.VALUE_200), htCallReceiptQueryBo.getBeginTime(), htCallReceiptQueryBo.getDeptId()).getRecords();
        if (CollUtil.isNotEmpty(callReportVoList)) {
            setCallReport(callReportVoList);
            CommonExcelUtil.exportExcel(callReportVoList.getFirst().getDeptName(), callReportVoList, GzxHtCallReportVo.class, response);
        } else {
            throw new CommonException("无可导出的数据");
        }
    }

    /**
     * 获取坐席工作报表
     *
     * @param htCallReceiptQueryBo
     * @return
     */
    @Override
    public Page<GzxHtCallReportVo> pageCallReportOld(GzxHtCallReceiptQueryBo htCallReceiptQueryBo) {
        if (htCallReceiptQueryBo.getDeptId() == null || StringUtils.isEmpty(htCallReceiptQueryBo.getBeginTime())) {
            throw new CommonException("时间和部门都不允许为空");
        }
        Page<GzxHtCallReportVo> callReportVoPage = htCallReceiptBusinessMapper.pageCallReportOld(new Page<>(htCallReceiptQueryBo.getCurrent(), htCallReceiptQueryBo.getSize()), htCallReceiptQueryBo.getBeginTime(), htCallReceiptQueryBo.getDeptId());
        if (CollectionUtil.isNotEmpty(callReportVoPage.getRecords())) {
            setCallReport(callReportVoPage.getRecords());
        }
        return callReportVoPage;
    }

    /**
     * 导出数据
     *
     * @param htCallReceiptQueryBo
     * @param response
     * @return
     */
    @Override
    public void exportCallReportOld(GzxHtCallReceiptQueryBo htCallReceiptQueryBo, HttpServletResponse response) {
        if (htCallReceiptQueryBo.getDeptId() == null || StringUtils.isEmpty(htCallReceiptQueryBo.getBeginTime())) {
            throw new CommonException("时间和部门都不允许为空");
        }
        List<GzxHtCallReportVo> callReportVoList = htCallReceiptBusinessMapper.pageCallReportOld(new Page<>(1, IntConstant.VALUE_200), htCallReceiptQueryBo.getBeginTime(), htCallReceiptQueryBo.getDeptId()).getRecords();
        if (CollectionUtil.isNotEmpty(callReportVoList)) {
            setCallReport(callReportVoList);
            CommonExcelUtil.exportExcel(callReportVoList.getFirst().getDeptName(), callReportVoList, GzxHtCallReportVo.class, response);
        } else {
            throw new CommonException("无可导出的数据");
        }
    }

    /**
     * 批量保存话单业务表
     *
     * @param consumerRecordList
     */
    @Override
    @DS("tjzy-async")
    @DSTransactional
    public void saveBatchBusiness(List<ConsumerRecord<String, String>> consumerRecordList) {
        TenantContext.ignoreHandle(() -> saveReceipt(getHtCallReceiptList(consumerRecordList)));
    }

    @Override
    @DS("tjzy-async")
    @DSTransactional(rollbackFor = Exception.class)
    public void generateWorkRecord() {
        //扫描尚未添加作业记录的话单业务表
        List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList = htCallReceiptBusinessMapper.selectList(Wrappers.<GzxHtCallReceiptBusiness>lambdaQuery()
                .select(GzxHtCallReceiptBusiness::getId, GzxHtCallReceiptBusiness::getCaseId, GzxHtCallReceiptBusiness::getCallee, GzxHtCallReceiptBusiness::getEndTime)
                .ge(GzxHtCallReceiptBusiness::getCreatedTime, CommonDateUtil.getTimeBeforeOrAfterMinutes(null, -30))
                .eq(GzxHtCallReceiptBusiness::getCallType, GzxHtCallTypeConstant.PREDICTIVE_CALL)
                .eq(GzxHtCallReceiptBusiness::getCallDuration, 0)
                .eq(GzxHtCallReceiptBusiness::getIsSaveWork, 0)
                .isNotNull(GzxHtCallReceiptBusiness::getCaseId)
                .last("limit 1000"));
        if (CollectionUtil.isNotEmpty(htCallReceiptBusinessList)) {
            List<GzxMediatorWork> mediatorWorkList = new ArrayList<>();
            //查询案件
            Set<Long> caseIdSet = htCallReceiptBusinessList.stream().map(GzxHtCallReceiptBusiness::getCaseId).collect(Collectors.toSet());
            List<GzxCase> caseList = caseDbService.selectCaseList(Wrappers.<GzxCase>lambdaQuery()
                    .select(GzxCase::getId, GzxCase::getDebtorCardId, GzxCase::getProjectCode, GzxCase::getMediatorId, GzxCase::getAffiliatedCompany, GzxCase::getTenantId)
                    .in(GzxCase::getId, caseIdSet));
            if (CollectionUtil.isNotEmpty(caseList)) {
                //根据身份证号查询三方联系人
                Set<String> debtorCardIdSet = caseList.stream().map(GzxCase::getDebtorCardId).map(CommonCryptogramUtil::doSm1AesEncrypt).collect(Collectors.toSet());
                List<GzxDebtorLiaison> debtorLiaisonList = debtorLiaisonService.selectList(Wrappers.<GzxDebtorLiaison>lambdaQuery()
                        .in(GzxDebtorLiaison::getDebtorCardId, debtorCardIdSet));
                Map<Long, GzxCase> caseMap = caseList.stream().collect(Collectors.toMap(GzxCase::getId, Function.identity()));
                Map<String, GzxDebtorLiaison> debtorLiaisonMap = debtorLiaisonList.stream().collect(Collectors.toMap(s -> s.getDebtorCardId() + s.getLiaisonPhone(), s -> s, (oldVal, newVal) -> newVal));
                List<Long> businessIdList = new ArrayList<>();
                for (GzxHtCallReceiptBusiness htCallReceiptBusiness : htCallReceiptBusinessList) {
                    if ((htCallReceiptBusiness.getCallDuration() == null || htCallReceiptBusiness.getCallDuration() == 0) && htCallReceiptBusiness.getCaseId() != null && caseMap.containsKey(htCallReceiptBusiness.getCaseId())) {
                        GzxCase gzxCase = caseMap.get(htCallReceiptBusiness.getCaseId());
                        String debtorCardId = gzxCase.getDebtorCardId();
                        GzxMediatorWork mediatorWork = new GzxMediatorWork();
                        mediatorWork.setCaseId(gzxCase.getId());
                        mediatorWork.setDebtorCardId(debtorCardId);
                        mediatorWork.setProjectCode(gzxCase.getProjectCode());
                        mediatorWork.setLiaisonPhone(htCallReceiptBusiness.getCallee());
                        mediatorWork.setWorkType(2);
                        mediatorWork.setWorkResult(2);
                        mediatorWork.setAffiliatedCompany(gzxCase.getAffiliatedCompany());
                        mediatorWork.setTenantId(gzxCase.getTenantId());
                        mediatorWork.setCreateUser(String.valueOf(gzxCase.getMediatorId()));
                        mediatorWork.setCreateTime(CommonDateUtil.strToDateTime(htCallReceiptBusiness.getEndTime()));
                        mediatorWork.setPhoneSource(-1);
                        String liaisonMapKey = debtorCardId + htCallReceiptBusiness.getCallee();
                        if (debtorLiaisonMap.containsKey(liaisonMapKey)) {
                            GzxDebtorLiaison debtorLiaison = debtorLiaisonMap.get(liaisonMapKey);
                            mediatorWork.setLiaisonName(debtorLiaison.getLiaisonName());
                            mediatorWork.setCallRecipient(debtorLiaison.getRelationship());
                            mediatorWork.setPhoneSource(debtorLiaison.getPhoneSource());
                        } else {
                            log.error("自动生成预测外呼记录，未匹配到联系人信息：{}", liaisonMapKey);
                        }
                        mediatorWorkList.add(mediatorWork);
                        businessIdList.add(htCallReceiptBusiness.getId());
                    }
                }
                mediatorWorkService.insertBatch(mediatorWorkList);
                if (CollectionUtil.isNotEmpty(businessIdList)) {
                    htCallReceiptBusinessMapper.update(Wrappers.<GzxHtCallReceiptBusiness>lambdaUpdate()
                            .set(GzxHtCallReceiptBusiness::getIsSaveWork, 1)
                            .in(GzxHtCallReceiptBusiness::getId, businessIdList));
                }
            }
        }
    }

    //定时任务发送邮件
    @Override
    @DS("tjzy-async")
    public void sendEmail(List<String> toAddress) {
        String excelPath = "";
        List<ExcelCalleeVo> list = htCallReceiptBusinessMapper.getCalleeList();
        //过滤掉callType不在1，2的数据
        list = list.stream().filter(item -> item.getCallType() == 1 || item.getCallType() == 2).toList();
        if (CollUtil.isNotEmpty(list)) {
            transService.transMore(list);
            //发邮件
            try {
                // 生成Excel
                excelPath = EmailExcelUtils.createExcel(list);
                wechatMailUtil.sendEmail(new MailRequestDto(toAddress, CommonDateUtil.getPreviousDayStr() + "手机号拨打次数大于5次的坐席", CommonDateUtil.getPreviousDayStr() + "手机号拨打次数大于5次的坐席", List.of(new File(excelPath))));
            } catch (MessagingException e) {
                log.error("定时任务发送邮件 - 邮件发送异常 ", e);
            } catch (Exception e) {
                log.error("定时任务发送邮件异常 ", e);
            } finally {
                if (StringUtils.isNotBlank(excelPath)) {
                    //删除临时文件 excelPath
                    File excelFile = new File(excelPath);
                    if (excelFile.exists() && excelFile.delete()) {
                        log.info("定时任务发送邮件 -临时文件删除成功");
                    }
                }
            }
        }
    }

    private void saveReceipt(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        Map<Integer, List<GzxHtCallReceiptBusiness>> callTypeMap = htCallReceiptBusinessList.stream().collect(Collectors.groupingBy(GzxHtCallReceiptBusiness::getCallType));
        for (Map.Entry<Integer, List<GzxHtCallReceiptBusiness>> entry : callTypeMap.entrySet()) {
            //来电
            if (GzxHtCallTypeConstant.INCOMING_CALL == entry.getKey()) {
                completeDataByIncoming(entry.getValue());
            }
            //单点外呼
            if (GzxHtCallTypeConstant.OUTGOING_CALL == entry.getKey()) {
                completeDataByOutGoing(entry.getValue());
            }
            //预测外呼
            if (GzxHtCallTypeConstant.PREDICTIVE_CALL == entry.getKey()) {
                completeDataByPredict(entry.getValue());
            }
        }
        //发送到质检Kafka
        sendQcKafka(htCallReceiptBusinessList);
        //批量保存到话单业务表
        insertBatchDb(htCallReceiptBusinessList);
    }

    private void setCallReport(List<GzxHtCallReportVo> callReportVoList) {
        Map<String, cn.hutool.json.JSONObject> userMap = userCaffeine.getUserMap();
        for (GzxHtCallReportVo callReportVo : callReportVoList) {
            callReportVo.setInfo(callReportVo.getConnectCalls() + "/" + callReportVo.getTotalCalls());
            callReportVo.setConnRate(CommonCalculateUtil.calculatePercentage(callReportVo.getConnectCalls(), callReportVo.getTotalCalls()));
            callReportVo.setNoValidConnected(callReportVo.getConnectCalls() - callReportVo.getValidConnected());
            callReportVo.setValidConnectRate(CommonCalculateUtil.calculatePercentage(callReportVo.getValidConnected(), callReportVo.getTotalCalls()));
            JSONObject jsonObject = userMap.get(callReportVo.getAgentNum());
            if (jsonObject != null) {
                callReportVo.setUserName(userMap.get(callReportVo.getAgentNum()).getStr("name"));
                callReportVo.setDeptName(userMap.get(callReportVo.getAgentNum()).getStr("orgName"));
            } else {
                callReportVo.setUserName("呼入");
            }
        }
    }

    private void completeDataByIncoming(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        Map<String, Map<String, String>> calleeMap = callInProperties.getMapByCallee();
        for (GzxHtCallReceiptBusiness htCallReceiptBusinesses : htCallReceiptBusinessList) {
            if (calleeMap.containsKey(htCallReceiptBusinesses.getCallee())) {
                htCallReceiptBusinesses.setTenantId(calleeMap.get(htCallReceiptBusinesses.getCallee()).get("tenantId"));
                htCallReceiptBusinesses.setDeptId(Long.valueOf(calleeMap.get(htCallReceiptBusinesses.getCallee()).get("deptId")));
            } else {
                htCallReceiptBusinesses.setTenantId(GzxTenantConstant.GZX_TENANT_ID);
            }
        }
    }

    private void completeDataByOutGoing(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        Map<String, GzxCase> caseMap = getCaseMap(htCallReceiptBusinessList);
        for (GzxHtCallReceiptBusiness htCallReceiptBusiness : htCallReceiptBusinessList) {
            //是案件，从案件表获取部门和租户
            if (StringUtils.isNotBlank(htCallReceiptBusiness.getBusiness()) && caseMap.containsKey(htCallReceiptBusiness.getBusiness())) {
                htCallReceiptBusiness.setCaseId(Long.valueOf(htCallReceiptBusiness.getBusiness()));
                GzxCase gzxCase = caseMap.get(htCallReceiptBusiness.getBusiness());
                htCallReceiptBusiness.setDeptId(gzxCase.getAffiliatedCompany());
                htCallReceiptBusiness.setTenantId(gzxCase.getTenantId());
            } else {
                //是租户ID
                Map<String, JSONObject> userMap = userCaffeine.getUserMapByTenantId(htCallReceiptBusiness.getBusiness());
                if (CollectionUtil.isNotEmpty(userMap) && userMap.containsKey(htCallReceiptBusiness.getAgentNum())) {
                    htCallReceiptBusiness.setTenantId(htCallReceiptBusiness.getBusiness());
                    htCallReceiptBusiness.setDeptId(userMap.get(htCallReceiptBusiness.getAgentNum()).getLong("orgId"));
                }
            }
        }
    }

    private void completeDataByPredict(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        List<Long> detailId = htCallReceiptBusinessList.stream().map(GzxHtCallReceiptBusiness::getBusiness).map(Long::parseLong).toList();
        List<GzxPredictCallDetail> predictCallDetailList = predictCallDbService.selectDetailList(Wrappers.<GzxPredictCallDetail>lambdaQuery()
                .select(GzxPredictCallDetail::getId, GzxPredictCallDetail::getCaseId, GzxPredictCallDetail::getTenantId, GzxPredictCallDetail::getCreateDept, GzxPredictCallDetail::getCreateUser)
                .in(GzxPredictCallDetail::getId, detailId));
        Map<String, GzxPredictCallDetail> predictCallDetailMap = predictCallDetailList.stream()
                .collect(Collectors.toMap(
                        detail -> String.valueOf(detail.getId()),
                        Function.identity()));
        for (GzxHtCallReceiptBusiness htCallReceiptBusiness : htCallReceiptBusinessList) {
            if (predictCallDetailMap.containsKey(htCallReceiptBusiness.getBusiness())) {
                GzxPredictCallDetail predictCallDetail = predictCallDetailMap.get(htCallReceiptBusiness.getBusiness());
                htCallReceiptBusiness.setTenantId(predictCallDetail.getTenantId());
                htCallReceiptBusiness.setDeptId(predictCallDetail.getCreateDept());
                htCallReceiptBusiness.setCaseId(predictCallDetail.getCaseId());
            } else {
                Map<String, JSONObject> userMap = userCaffeine.getUserMapByTenantId(GzxTenantConstant.GZX_TENANT_ID);
                if (CollectionUtil.isNotEmpty(userMap) && userMap.containsKey(htCallReceiptBusiness.getAgentNum())) {
                    htCallReceiptBusiness.setTenantId(htCallReceiptBusiness.getBusiness());
                    htCallReceiptBusiness.setDeptId(userMap.get(htCallReceiptBusiness.getAgentNum()).getLong("orgId"));
                }
            }
        }
    }

    private Map<String, GzxCase> getCaseMap(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        Set<Long> businessSet = htCallReceiptBusinessList.stream()
                .filter(s -> StringUtils.isNotBlank(s.getBusiness()))
                .map(s -> Long.valueOf(s.getBusiness()))
                .collect(Collectors.toSet());
        return caseDbService.selectCaseList(
                        Wrappers.<GzxCase>lambdaQuery()
                                .select(GzxCase::getId, GzxCase::getAffiliatedCompany, GzxCase::getTenantId)
                                .in(GzxCase::getId, businessSet))
                .stream()
                .collect(Collectors.toMap(
                        caseObj -> String.valueOf(caseObj.getId()),  // 将Long类型的ID转为String
                        Function.identity(),
                        (oldVal, newVal) -> newVal));
    }

    private void sendQcKafka(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        Map<String, cn.hutool.json.JSONObject> userMap = userCaffeine.getUserMapByTenantId(GzxTenantConstant.GZX_TENANT_ID);
        for (GzxHtCallReceiptBusiness htCallReceiptBusinesses : htCallReceiptBusinessList) {
            if (htCallReceiptBusinesses.getCallDuration() > 10) {
                GzxCallCheckDto callCheckDto = new GzxCallCheckDto();
                callCheckDto.setCallId(htCallReceiptBusinesses.getCallId());
                callCheckDto.setCallType(htCallReceiptBusinesses.getCallType());
                callCheckDto.setAgentNum(htCallReceiptBusinesses.getAgentNum());
                if (StringUtils.isNotBlank(htCallReceiptBusinesses.getAgentNum()) && userMap.containsKey(htCallReceiptBusinesses.getAgentNum())) {
                    callCheckDto.setUserName(userMap.get(htCallReceiptBusinesses.getAgentNum()).getStr("name"));
                }
                callCheckDto.setCallee(htCallReceiptBusinesses.getCallee());
                callCheckDto.setCallDuration(htCallReceiptBusinesses.getCallDuration());
                callCheckDto.setFileUrl(htCallReceiptBusinesses.getFileUrl());
                callCheckDto.setDeptId(htCallReceiptBusinesses.getDeptId());
                kafkaMessageProducer.sendMessage(KafkaTopicConstant.QC_CALL_RECEIPT_TOPIC, htCallReceiptBusinesses.getCallId(), JSON.toJSONString(callCheckDto));
            }
        }
    }

    private void insertBatchDb(List<GzxHtCallReceiptBusiness> htCallReceiptBusinessList) {
        List<List<GzxHtCallReceiptBusiness>> batchList = Lists.partition(htCallReceiptBusinessList, IntConstant.VALUE_1000);
        String currentDate = CommonDateUtil.getCurrentTimeStr();
        for (List<GzxHtCallReceiptBusiness> batch : batchList) {
            String sql = batch.stream()
                    .map(s -> String.format("('%s','%s', '%s', '%s', '%s','%s','%s','%s','%s','%s','%s','%s', '%s', '%s', '%s','%s','%s','%s','%s','%s','%s','%s', '%s', '%s', '%s','%s','%s','%s','%s','%s','%s','%s')",
                            escapeSql(IdUtil.getSnowflakeNextIdStr()),
                            escapeSql(s.getBusiness()),
                            escapeSql(String.valueOf(s.getCallType())),
                            escapeSql(s.getCallId()),
                            escapeSql(s.getAgentNum()),
                            escapeSql(s.getUserName()),
                            escapeSql(s.getCaller()),
                            escapeSql(s.getCallee()),
                            escapeSql(s.getRingTime()),
                            escapeSql(String.valueOf(s.getRingDuration())),
                            escapeSql(s.getConnectTime()),
                            escapeSql(s.getStartTime()),
                            escapeSql(s.getEndTime()),
                            escapeSql(String.valueOf(s.getCallDuration())),
                            escapeSql(s.getFileUrl()),
                            escapeSql(s.getArea()),
                            escapeSql(s.getResult()),
                            escapeSql(String.valueOf(s.getSignal())),
                            escapeSql(s.getTaskId()),
                            escapeSql(s.getImportTime()),
                            escapeSql(s.getAddTime()),
                            escapeSql(s.getAcdEndTime()),
                            escapeSql(String.valueOf(s.getAcdDuration())),
                            escapeSql(String.valueOf(s.getInviteComments())),
                            escapeSql(s.getMydKey()),
                            escapeSql(String.valueOf(s.getAsrCode())),
                            escapeSql(s.getAsrName()),
                            escapeSql(s.getCompanyName()),
                            escapeSql(String.valueOf(s.getCaseId())),
                            escapeSql(s.getTenantId()),
                            escapeSql(String.valueOf(s.getDeptId())),
                            escapeSql(currentDate)))
                    .collect(Collectors.joining(", "));
            htCallReceiptBusinessMapper.insertBatchByValSql(sql.replace("'null'", "null"));
            CommonSleepUtil.sleep(IntConstant.VALUE_100);
        }
    }

    private List<GzxHtCallReceiptBusiness> getHtCallReceiptList(List<ConsumerRecord<String, String>> recordList) {
        return recordList.stream().map(rowData -> JSON.parseObject(rowData.value(), GzxHtCallReceiptBusiness.class)).toList();
    }
}
